package tsuna

import (
	"context"
	"fmt"
	"io"

	"github.com/jeanphorn/log4go"
	"github.com/tsuna/gohbase"
	"github.com/tsuna/gohbase/filter"
	"github.com/tsuna/gohbase/hrpc"
	"github.com/tsuna/gohbase/pb"
)

var _client gohbase.Client

//connect
func ConnectHBase(zooRoot string) {
	_client = gohbase.NewClient(zooRoot)
}

//Insert a cell
func Insert(table string, row string, cf string, col string, val string) (*hrpc.Result, error) {
	// Values maps a ColumnFamily -> Qualifiers -> Values.
	values := map[string]map[string][]byte{cf: map[string][]byte{col: []byte(val)}}
	//fmt.Println(values)
	putRequest, err := hrpc.NewPutStr(context.Background(), table, row, values)
	rsp, err := _client.Put(putRequest)
	return rsp, err
}

//Get a specific cell with a filter
func GetRow(table string, row string) (*hrpc.Result, error) {
	// Perform a get for the cell with key "15", column family "cf" and qualifier "a"
	getRequest, err := hrpc.NewGetStr(context.Background(), table, row)
	getRsp, err := _client.Get(getRequest)
	return getRsp, err
}

//Get a specific cell with a filter
func Getcell(table string, row string, cf string, col string) (*hrpc.Result, error) {
	// Perform a get for the cell with key "15", column family "cf" and qualifier "a"
	family := map[string][]string{cf: []string{col}}
	getRequest, err := hrpc.NewGetStr(context.Background(), table, row,
		hrpc.Families(family))
	getRsp, err := _client.Get(getRequest)
	return getRsp, err
}

//Get a specific cell with a filter

func Getscount(table string, startrow string, endrow string) (int, error) {

	// Perform a get for the cell with key "15", column family "cf" and qualifier "a"
	//pFilter1 := filter.NewPageFilter(NewPageFilter + 1)
	//pFilter2 := filter.NewRowRange([]byte(startrow), []byte(endrow), true, true)
	//pFilter3 := filter.NewFirstKeyOnlyFilter()

	getRequest, _ := hrpc.NewScanRangeStr(context.Background(), table, startrow, endrow /*, hrpc.Filters(pFilter1) , hrpc.Filters(pFilter3)*/)
	scan := _client.Scan(getRequest)
	var res []*hrpc.Result
	var err error
	for {
		getRsp, err := scan.Next()
		if err == io.EOF || getRsp == nil {
			//fmt.Println("err == io.EOF || getRsp == nil")
			break
		}
		if err != nil {
			//fmt.Println("getscan:error")
		} else {
			res = append(res, getRsp)
		}
	}
	return len(res), err

}

//Get a specific cell with a filter
func Getscan(table string, startrow string, endrow string) ([]*hrpc.Result, error) {
	// Perform a get for the cell with key "15", column family "cf" and qualifier "a"
	pFilter2 := filter.NewInclusiveStopFilter([]byte(endrow))
	getRequest, _ := hrpc.NewScanRangeStr(context.Background(), table, startrow, endrow, hrpc.Filters(pFilter2))
	scan := _client.Scan(getRequest)
	var res []*hrpc.Result
	var err error
	for {
		getRsp, err := scan.Next()
		if err == io.EOF || getRsp == nil {
			//fmt.Println("err == io.EOF || getRsp == nil")
			break
		}
		if err != nil {
			//fmt.Println("getscan:error")
		} else {
			res = append(res, getRsp)
		}
	}
	return res, err
}

//Get a specific cell with a filter
func GetscanWithPrefix(table string, startrow string, endrow string, Prefix string) ([]*hrpc.Result, error) {
	// Perform a get for the cell with key "15", column family "cf" and qualifier "a"
	pFilter1 := filter.NewPrefixFilter([]byte(Prefix))
	pFilter2 := filter.NewInclusiveStopFilter([]byte(endrow))
	getRequest, _ := hrpc.NewScanRangeStr(context.Background(), table, startrow, endrow,
		hrpc.Filters(pFilter1), hrpc.Filters(pFilter2))
	scan := _client.Scan(getRequest)
	var res []*hrpc.Result
	var err error
	for {
		getRsp, err := scan.Next()
		if err == io.EOF || getRsp == nil {
			//fmt.Println("err == io.EOF || getRsp == nil")
			break
		}
		if err != nil {
			//fmt.Println("getscan:error")
		} else {
			res = append(res, getRsp)
		}
	}
	return res, err
}

//Get a specific cell with a filter
func GetscanWithcol_1(table string, startrow string, endrow string, cf string,
	col string) ([]*hrpc.Result, error) {

	// Perform a get for the cell with key "15", column family "cf" and qualifier "a"
	family1 := map[string][]string{cf: []string{col}}

	getRequest, _ := hrpc.NewScanRangeStr(context.Background(), table, startrow, endrow,
		hrpc.Families(family1))
	scan := _client.Scan(getRequest)
	var res []*hrpc.Result
	var err error
	for {
		getRsp, err := scan.Next()
		if err == io.EOF || getRsp == nil {
			//fmt.Println("err == io.EOF || getRsp == nil")
			break
		}
		if err != nil {
			//fmt.Println("getscan:error")
		} else {
			res = append(res, getRsp)
		}
	}
	return res, err
}

//Get a specific cell with a filter
func GetscanWithcol_2(table string, startrow string, endrow string, cf string,
	col_1 string, col_2 string) ([]*hrpc.Result, error) {

	// Perform a get for the cell with key "15", column family "cf" and qualifier "a"
	family1 := map[string][]string{cf: []string{col_1, col_2}}
	//family2 := map[string][]string{cf: []string{col_2}}
	getRequest, _ := hrpc.NewScanRangeStr(context.Background(), table, startrow, endrow,
		hrpc.Families(family1) /*, hrpc.Families(family2)*/)
	scan := _client.Scan(getRequest)
	var res []*hrpc.Result
	var err error
	for {
		getRsp, err := scan.Next()
		if err == io.EOF || getRsp == nil {
			//fmt.Println("err == io.EOF || getRsp == nil")
			break
		}
		if err != nil {
			//fmt.Println("getscan:error")
		} else {
			res = append(res, getRsp)
		}
	}
	return res, err
}

//Get a specific cell with a filter
func GetscanWithpage(table string, startrow string, endrow string, NewPageFilter int64) ([]*hrpc.Result, string, error) {
	// Perform a get for the cell with key "15", column family "cf" and qualifier "a"

	pFilter1 := filter.NewPageFilter(NewPageFilter + 1)
	pFilter2 := filter.NewInclusiveStopFilter([]byte(endrow))
	getRequest, _ := hrpc.NewScanRangeStr(context.Background(), table, startrow, endrow, hrpc.Filters(pFilter1), hrpc.Filters(pFilter2))
	scan := _client.Scan(getRequest)
	var res []*hrpc.Result
	var err error
	for {
		getRsp, err := scan.Next()
		if err == io.EOF || getRsp == nil {
			//fmt.Println("err == io.EOF || getRsp == nil")
			break
		}
		if err != nil {
			//fmt.Println("getscan:error")
		} else {
			if int64(len(res)) < NewPageFilter {
				res = append(res, getRsp)
			} else {
				indexrow := string((*pb.Cell)(getRsp.Cells[0]).GetRow())
				return res, indexrow, err
			}
		}
	}
	return res, "", err
}

//Get a specific cell with a filter
func GetscanWithpagecol_1(table string, startrow string, endrow string, cf string, col string, NewPageFilter int64) ([]*hrpc.Result, string, error) {

	// Perform a get for the cell with key "15", column family "cf" and qualifier "a"

	pFilter1 := filter.NewPageFilter(NewPageFilter + 1)
	pFilter2 := filter.NewInclusiveStopFilter([]byte(endrow))
	family1 := map[string][]string{cf: []string{col}}

	getRequest, _ := hrpc.NewScanRangeStr(context.Background(), table, startrow, endrow,
		hrpc.Filters(pFilter1), hrpc.Filters(pFilter2), hrpc.Families(family1))
	scan := _client.Scan(getRequest)
	var res []*hrpc.Result
	var err error
	for {
		getRsp, err := scan.Next()
		if err == io.EOF || getRsp == nil {
			//fmt.Println("err == io.EOF || getRsp == nil")
			break
		}
		if err != nil {
			//fmt.Println("getscan:error")
		} else {
			if int64(len(res)) < NewPageFilter {
				res = append(res, getRsp)
			} else {
				indexrow := string((*pb.Cell)(getRsp.Cells[0]).GetRow())
				return res, indexrow, err
			}
		}
	}
	return res, "", err
}

//Get a specific cell with a filter
func PrintRes(getres *hrpc.Result, err error) {
	if err != nil {
		log4go.Error("PrintRes: %s", err.Error())
		return
	}

	log4go.Info("PrintRes: %s", "方式1")
	for i, v := range getres.Cells {
		row := string(v.Row[:])
		fam := string(v.Family[:])
		qua := string(v.Qualifier[:])
		value := string(v.Value[:])
		str := fmt.Sprintf("index:%d Rowkey: %s	family: %s qualifies:%s	value:%s", i, row, fam, qua, value)
		fmt.Println(str)
	}

	log4go.Info("PrintRes: %s", "方式2")
	for _, cell := range getres.Cells {
		fmt.Println(string((*pb.Cell)(cell).GetRow()))
		fmt.Println(string((*pb.Cell)(cell).GetFamily()))
		fmt.Println(string((*pb.Cell)(cell).GetQualifier()))
		fmt.Println(string((*pb.Cell)(cell).GetValue()))
		fmt.Println((*pb.Cell)(cell).GetTimestamp())

	}
}

//Get a specific cell with a filter
func PrintScanWithcol(getres []*hrpc.Result, err error) {
	if err != nil {
		log4go.Error("PrintScan: %s", err.Error())
		return
	}

	log4go.Info("PrintRes: %s", "方式1")
	var k int
	for k = 0; k < len(getres); k++ {
		for i, v := range getres[k].Cells {
			row := string(v.Row[:])
			fam := string(v.Family[:])
			qua := string(v.Qualifier[:])
			value := string(v.Value[:])
			str := fmt.Sprintf("index:%d Rowkey: %s	family: %s qualifies:%s	value:%s", i, row, fam, qua, value)
			fmt.Println(str)
		}
	}
}

//Get a specific cell with a filter
func PrintScan(getres []*hrpc.Result, row string, err error) {
	if err != nil {
		log4go.Error("PrintScan: %s", err.Error())
		return
	}

	log4go.Info("PrintRes: %s", "方式1")
	var k int
	for k = 0; k < len(getres); k++ {
		for i, v := range getres[k].Cells {
			row := string(v.Row[:])
			fam := string(v.Family[:])
			qua := string(v.Qualifier[:])
			value := string(v.Value[:])
			str := fmt.Sprintf("index:%d Rowkey: %s	family: %s qualifies:%s	value:%s", i, row, fam, qua, value)
			fmt.Println(str)
		}
	}
	log4go.Info("PrintRes: %s", "方式2")
	for k = 0; k < len(getres); k++ {
		for _, cell := range getres[k].Cells {
			fmt.Println(string((*pb.Cell)(cell).GetRow()))
			fmt.Println(string((*pb.Cell)(cell).GetFamily()))
			fmt.Println(string((*pb.Cell)(cell).GetQualifier()))
			fmt.Println(string((*pb.Cell)(cell).GetValue()))
			fmt.Println((*pb.Cell)(cell).GetTimestamp())
		}
	}
}
